package com.ia.aistream.manager.framework.redis.impl;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.ia.aistream.manager.framework.redis.RedisManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.core.*;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 *
 */
@Slf4j
@Service
public class RedisManagerImpl implements RedisManager {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Override
    public Boolean exists(String key) {
        return redisTemplate.hasKey(key);
    }

    @Override
    public void del(String key, String... keys) {
        List<String> list;
        if (keys == null || keys.length == 0) {
            list = Lists.newArrayList();
        } else {
            list = Lists.newArrayList(keys);
        }

        list.add(key);
        redisTemplate.delete(list);
    }

    @Override
    public Set<String> keys(String pattern) {
        return redisTemplate.keys(pattern);
    }

    @Override
    public void expire(String key, long timeout) {
        redisTemplate.expire(key, timeout, TimeUnit.SECONDS);
    }

    @Override
    public void expire(String key, long timeout, TimeUnit unit) {
        redisTemplate.expire(key, timeout, unit);
    }

    @Override
    public void multi() {
        redisTemplate.multi();
    }

    @Override
    public void exec() {
        redisTemplate.exec();
    }

    @Override
    public void discard() {
        redisTemplate.discard();
    }

    @Override
    public String lock(String key, int timeout) {
        int retryTime = timeout * 1000;
        try {
            while (retryTime > 0) {
                Long redisTime = redisTemplate.execute(RedisServerCommands::time);
                if (redisTime == null) {
                    retryTime -= 100;
                    Thread.sleep(100);
                    continue;
                }

                Boolean flag = redisTemplate.execute((RedisCallback<Boolean>) connection -> {
                    Boolean result = connection.setNX(key.getBytes(), String.valueOf(redisTime + timeout * 1000).getBytes());
                    if (result != null && result) {
                        connection.expire(key.getBytes(), timeout);
                    }
                    return result;
                });

                if (flag != null && flag) {
                    //设置超时时间，释放内存
                    return String.valueOf(redisTime + timeout * 1000);
                }

                String result = redisTemplate.opsForValue().get(key);
                if (!StringUtils.isNumeric(result)) {
                    redisTemplate.delete(key);
                    continue;
                }

                long cacheTime = Long.parseLong(result);
                //锁已经失效
                if (cacheTime < redisTime) {
                    //判断是否为空，不为空时，说明已经失效，如果被其他线程设置了值，则第二个条件判断无法执行
                    //获取上一个锁到期时间，并设置现在的锁到期时间
                    redisTemplate.opsForValue().setIfAbsent(key, String.valueOf(redisTime + timeout * 1000));
                    redisTemplate.expire(key, timeout, TimeUnit.SECONDS);
                    return String.valueOf(redisTime + timeout * 1000);
                }

                retryTime -= 200;
                Thread.sleep(100);
            }
        } catch (Exception e) {
            log.error("[ redis锁报错 ] , key:{}, waitTime:5000, timeout:{}", key, timeout, e);
        }
        return null;
    }

    @Override
    public void unlock(String key, String lock) {
        //获取redis中设置的时间
        String result = redisTemplate.opsForValue().get(key);
        //如果是加锁者，则删除锁， 如果不是，则等待自动过期，重新竞争加锁
        if (Objects.equals(result, lock)) {
            redisTemplate.delete(lock);
        }
    }

    @Override
    public void set(String key, Object value) {
        redisTemplate.opsForValue().set(key, JSON.toJSONString(value));
    }

    @Override
    public Boolean setnx(String key, Object value) {
        return redisTemplate.opsForValue().setIfAbsent(key, JSON.toJSONString(value));
    }

    @Override
    public void setex(String key, int seconds, Object value) {
        redisTemplate.opsForValue().set(key, JSON.toJSONString(value), seconds, TimeUnit.SECONDS);
    }

    @Override
    public void psetex(String key, long milliseconds, Object value) {
        redisTemplate.opsForValue().setIfPresent(key, JSON.toJSONString(value), milliseconds, TimeUnit.MILLISECONDS);
    }

    @Override
    public <T> T get(String key, Class<T> clazz) {
        String value = redisTemplate.opsForValue().get(key);
        if (StringUtils.isBlank(value)) {
            return null;
        }
        return JSON.parseObject(value, clazz);
    }

    @Override
    public <T> T getSet(String key, T value) {
        String old = redisTemplate.opsForValue().getAndSet(key, JSON.toJSONString(value));
        return JSON.parseObject(old, value.getClass().asSubclass(value.getClass()));
    }

    @Override
    public Long strLen(String key) {
        return redisTemplate.opsForValue().size(key);
    }

    @Override
    public Long incr(String key) {
        return redisTemplate.opsForValue().increment(key, 1);
    }

    @Override
    public void incrby(String key, long increment) {
        redisTemplate.opsForValue().increment(key, increment);
    }

    @Override
    public void incrbyfloat(String key, double increment) {
        redisTemplate.opsForValue().increment(key, increment);
    }

    @Override
    public <T> List<T> mget(Collection<String> keys, Class<T> clazz) {
        List<String> list = redisTemplate.opsForValue().multiGet(keys);
        if (CollectionUtils.isEmpty(list)) {
            return Lists.newArrayList();
        }


        List<T> result = Lists.newArrayList();
        list.forEach(value -> result.add(JSON.parseObject(value, clazz)));
        return result;
    }


    @Override
    public void hset(String key, String field, Object value) {
        redisTemplate.opsForHash().put(key, field, JSON.toJSONString(value));
    }

    @Override
    public boolean hsetnx(String key, String field, Object value) {
        return redisTemplate.opsForHash().putIfAbsent(key, field, JSON.toJSONString(value));
    }

    @Override
    public <V> V hget(String key, String field, Class<V> clazz) {
        BoundHashOperations<String, String, String> operations = redisTemplate.boundHashOps(key);
        String value = operations.get(field);
        if (org.apache.commons.lang3.StringUtils.isEmpty(value)) {
            return null;
        }

        return JSON.parseObject(value, clazz);
    }

    @Override
    public boolean hexists(String key, String field) {
        return redisTemplate.opsForHash().hasKey(key, field);
    }

    @Override
    public void hdel(String key, Object... fields) {
        redisTemplate.opsForHash().delete(key, fields);
    }

    @Override
    public Long hlen(String key) {
        return redisTemplate.opsForHash().size(key);
    }

    @Override
    public Long hstrlen(String key, String field) {
        return redisTemplate.opsForHash().lengthOfValue(key, field);
    }

    @Override
    public Long hincrby(String key, String field, long increment) {
        return redisTemplate.opsForHash().increment(key, field, increment);
    }

    @Override
    public Double hincrbyfloat(String key, String field, double increment) {
        return redisTemplate.opsForHash().increment(key, field, increment);
    }

    @Override
    public <V> void hmset(String key, Map<String, V> map) {
        Map<String, String> param = Maps.newHashMap();
        map.forEach((field, value) -> param.put(field, JSON.toJSONString(value)));
        redisTemplate.opsForHash().putAll(key, param);
    }

    @Override
    public <V> List<V> hmget(String key, Collection<String> fields, Class<V> clazz) {
        BoundHashOperations<String, String, String> operations = redisTemplate.boundHashOps(key);
        List<String> list = operations.multiGet(fields);
        if (org.apache.commons.collections.CollectionUtils.isEmpty(list)) {
            return Lists.newArrayList();
        }

        List<V> result = Lists.newArrayList();
        list.forEach(element -> result.add(JSON.parseObject(element, clazz)));
        return result;
    }

    @Override
    public Set<String> hkeys(String key) {
        BoundHashOperations<String, String, String> operations = redisTemplate.boundHashOps(key);
        return operations.keys();
    }


    @Override
    public <V> List<V> hvals(String key, Class<V> clazz) {
        BoundHashOperations<String, String, String> operations = redisTemplate.boundHashOps(key);
        List<String> list = operations.values();
        if (org.apache.commons.collections.CollectionUtils.isEmpty(list)) {
            return Lists.newArrayList();
        }

        List<V> result = Lists.newArrayList();
        list.forEach(element -> {
            result.add(JSON.parseObject(element, clazz));
        });
        return result;
    }


    @Override
    public <V> Map<String, V> hgetall(String key, Class<V> clazz) {
        BoundHashOperations<String, String, String> operations = redisTemplate.boundHashOps(key);
        Map<String, String> map = operations.entries();
        if (CollectionUtils.isEmpty(map)) {
            return Maps.newHashMap();
        }
        Map<String, V> result = Maps.newHashMap();
        map.forEach((field, value) -> result.put(field, JSON.parseObject(value, clazz)));
        return result;
    }

    @Override
    public void sadd(String key, Object... members) {
        if (members == null || members.length == 0) {
            return;
        }

        for (Object member : members) {
            redisTemplate.opsForSet().add(key, JSON.toJSONString(member));
        }
    }

    @Override
    public boolean sismembers(String key, Object member) {
        return Boolean.TRUE.equals(redisTemplate.opsForSet().isMember(key, JSON.toJSONString(member)));
    }


    @Override
    public <T> Set<T> smembers(String key, Class<T> clazz) {
        Set<String> set = redisTemplate.opsForSet().members(key);
        if (CollectionUtils.isEmpty(set)) {
            return Sets.newLinkedHashSet();
        }

        Set<T> result = Sets.newLinkedHashSet();
        set.forEach(menber -> result.add(JSON.parseObject(menber, clazz)));
        return result;
    }

    @Override
    public void srem(String key, Object member) {
        redisTemplate.opsForSet().remove(key, JSON.toJSONString(member));
    }

    @Override
    public void smove(String source, String destination, Object member) {
        redisTemplate.opsForSet().move(source, JSON.toJSONString(member), destination);
    }

    @Override
    public long scard(String key) {
        Long size = redisTemplate.opsForSet().size(key);
        return size == null ? 0 : size;
    }

    @Override
    public void zadd(String key, double score, Object member) {
        redisTemplate.boundZSetOps(key).add(JSON.toJSONString(member), score);
    }

    @Override
    public Double zscore(String key, Object member) {
        return redisTemplate.boundZSetOps(key).score(JSON.toJSONString(member));
    }

    @Override
    public Double zincrby(String key, double increment, Object member) {
        return redisTemplate.boundZSetOps(key).incrementScore(JSON.toJSONString(member), increment);
    }

    @Override
    public long zcard(String key) {
        Long card = redisTemplate.boundZSetOps(key).zCard();
        return card == null ? 0 : card;
    }


    @Override
    public <T> Set<T> zrange(String key, int start, int stop, Class<T> clazz) {
        Set<String> set = redisTemplate.boundZSetOps(key).range(start, stop);
        Set<T> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        set.forEach(string -> result.add(JSON.parseObject(string, clazz)));
        return result;
    }


    @Override
    public <T> Set<ZSetOperations.TypedTuple<T>> zrange(String key, int start, int stop, boolean withscores, Class<T> clazz) {
        Set<ZSetOperations.TypedTuple<String>> set = redisTemplate.boundZSetOps(key).rangeWithScores(start, stop);
        Set<ZSetOperations.TypedTuple<T>> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        set.forEach(tuple -> result.add(new DefaultTypedTuple<T>(JSON.parseObject(tuple.getValue(), clazz), withscores ? tuple.getScore() : null)));
        return result;
    }


    @Override
    public <T> Set<T> zrangebyscore(String key, double min, double max, Class<T> clazz) {
        Set<String> set = redisTemplate.opsForZSet().rangeByScore(key, min, max);
        Set<T> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        set.forEach(string -> result.add(JSON.parseObject(string, clazz)));
        return result;
    }


    @Override
    public <T> Set<T> zrangebyscore(String key, double min, double max, int offset, int count, Class<T> clazz) {
        Set<String> set = redisTemplate.opsForZSet().rangeByScore(key, min, max, offset, count);
        Set<T> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        set.forEach(string -> result.add(JSON.parseObject(string, clazz)));
        return result;
    }


    @Override
    public <T> Set<ZSetOperations.TypedTuple<T>> zrangebyscore(String key, double min, double max, boolean withscores, int offset, int count, Class<T> clazz) {
        Set<ZSetOperations.TypedTuple<T>> result = Sets.newLinkedHashSet();

        if (withscores) {
            Set<ZSetOperations.TypedTuple<String>> set = redisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max, offset, count);
            if (CollectionUtils.isEmpty(set)) {
                return result;
            }

            set.forEach(tuple -> result.add(new DefaultTypedTuple<T>(JSON.parseObject(tuple.getValue(), clazz), tuple.getScore())));
        } else {
            Set<String> set = redisTemplate.opsForZSet().rangeByScore(key, min, max, offset, count);
            if (CollectionUtils.isEmpty(set)) {
                return result;
            }
            set.forEach(string -> result.add(new DefaultTypedTuple<>(JSON.parseObject(string, clazz), null)));
        }

        return result;
    }


    @Override
    public <T> Set<T> zrevrange(String key, int start, int stop, Class<T> clazz) {
        Set<String> set = redisTemplate.boundZSetOps(key).reverseRange(start, stop);
        Set<T> result = Sets.newLinkedHashSet();
        if (CollectionUtils.isEmpty(set)) {
            return result;
        }

        set.forEach(string -> result.add(JSON.parseObject(string, clazz)));
        return result;
    }

    @Override
    public void zrem(String key, Object member) {
        redisTemplate.boundZSetOps(key).remove(JSON.toJSONString(member));
    }
}
